<?php
/**
 * worker：消费者
 * 从队列中取出消息并post
 * @Author: Cqiu
 * @Date: 2017-6-6
 */

define('APP_ROOT', __DIR__);

$continueAsk = true;

$config = include APP_ROOT . '/config/config.php';

echo '======Post Server Start====' . PHP_EOL;

$redis = new Redis();
$redis->connect($config['redis_host'], $config['redis_port']) or die('Redis connect failed!');
if (isset($config['redis_pass']))$redis->auth($config['redis_pass']) or die('Redis auth failed!');
$pid = getmypid();

echo 'Redis Connect! pid:' . $pid;

if ($redis->incr('cqiu_test_' . $pid)) {
    $redis->decr('cqiu_test_' . $pid);
} else {
    die('Redis write failed!');
}

//resetData
if (isset($argv[1])) {

    $tmp_queues = $redis->keys('tmp_queue*');
    echo 'reset tmp_queue', "\n";
    print_r($tmp_queues);
    $ids = trim(fgets(STDIN));
    foreach (preg_split('#\s+#', $ids) as $id) {
        while ($redis->rpoplpush('tmp_queue' . $id, 'net_car_queue')) ;
    }

    $tmp_queues = $redis->keys('net_car_queue_*');
    print_r($tmp_queues);
    echo "\n", 'reset net_car_queue_[code] (default all)>';
    $ids = trim(fgets(STDIN));
    if ($ids == '') {
        $ids = &$tmp_queues;
    } else {
        if ($ids[0] == '-') {
            $ids = preg_split('#\s+#', substr($ids, 1));
            foreach ($tmp_queues as $i => $queueName) {
                if (in_array($i, $ids)) {
                    unset($tmp_queues[$i]);
                }
            }
        } else {
            $ids = preg_split('#\s+#', $ids);
            foreach ($tmp_queues as $i => $queueName) {
                if (!in_array($i, $ids)) {
                    unset($tmp_queues[$i]);
                }
            }
        }
        $ids = $tmp_queues;
    }
    foreach ($ids as $id) {
        while ($redis->rpoplpush('net_car_queue_' . $id, 'net_car_queue')) ;
    }

    exit('=======done!' . "\n");
}

$apis = include APP_ROOT . '/config/net-car-apis.php';

defined('CURLOPT_COOKIE') or define('CURLOPT_COOKIE', 10022);

/**
 * 照重庆运管局的错误拼写来
 * @param string $IPCType
 * @return string
 */
function fixURLs($IPCType)
{
    if (in_array($IPCType, [
        'baseInfoDriverEducate',
        'baseInfoDriverApp',
        'baseInfoDriverStat',
        'baseInfoPassenger'
    ])) {
        return str_replace('baseInfo', 'baselnfo', $IPCType);
    }
    return $IPCType;
}

while (1) {

    $msg = $redis->rpoplpush('net_car_queue', 'tmp_queue' . $pid);

    if (!$msg) {
        continue;
    }

    _log($msg);
    $data = json_decode($msg, true);
    if (!$data) {
        $redis->lRem('tmp_queue' . $pid, $msg, 1);
        $redis->incr('invalid');
        continue;
    }

    //_log($data);
    if (isset($data['IPCType'])) {
        $IPCType = $data['IPCType'];

        if (isset($apis[$data['IPCType']])) {
            $apiInfo = $apis[$IPCType];
            $IPCType = fixURLs($IPCType);
            $data['IPCType'] = $IPCType;
            $data = preparePostData($apiInfo, $data);

            if (isset($data['CompanyId'])) {
                //_log($config['api_gateway'] . $apiInfo['url'] . '?company=syjcx');
                _log(json_encode($data, JSON_UNESCAPED_UNICODE));

                $resultCode = post(
//                    $config['api_gateway'] . $apiInfo['url'] . '?company=syjcx',
                    $config['api_gateway'] . $IPCType,
                    $data,
                    ['IPCType:' . $IPCType]);
            } else {
                $resultCode = $data;//error info
            }
        } elseif ($IPCType == 'file') {
            $resultCode = upfile($config['ftp'], $data['file'], $data['upPath']);
        } else {
            $resultCode = 0;
        }
        _log($resultCode);

        if ($resultCode == 201) {
            $redis->lRem('tmp_queue' . $pid, $msg, 1);
            $redis->incr('done');
        } else {
//            $redis->rpoplpush('tmp_queue' . $pid, 'net_car_queue');
//            $redis->incr('failed');
            if (!is_int($resultCode)) $resultCode = '000';
            $redis->rpoplpush('tmp_queue' . $pid, 'net_car_queue_' . $resultCode);
            $redis->incr('failed');
        }

    }

    if ($continueAsk) {
        if ($continueAsk > 1) {
            sleep($continueAsk);
        } else {
            echo "\n", str_repeat('_', 50), 'continue? > ';
            $continue = trim(fgets(STDIN));
            if ($continue == 'exit') exit;
            elseif ($continue == 'yes') $continueAsk = false;
            elseif (is_int($continue)) $continueAsk = $continue;
        }
    }

}


/**
 * 准备 数据
 * @param $apiInfo
 * @param $data
 * @return array
 */
function preparePostData($apiInfo, $data)
{
    $IPCType = $data['IPCType'];
    unset($data['IPCType']);
    $data['CompanyId'] = 'syjcx';
    if (isset($data['State']) && $apiInfo['fields']['State'][1] == '数字型') $data['State'] = 0;
    if (isset($data['Flag'])) $data['Flag'] = 1;

    $errors = [];
    foreach ($apiInfo['fields'] as $field => $info) {
        if ($info[0] == '是' && !isset($data[$field])) {
            $errors[] = "{$field}: required!";
            continue;
        }
        //处理函数！
        if (isset($info['handle']) && function_exists($info['handle'])) {
            $data[$field] = $info['handle']($data[$field]);
        }
        if ($info[1] == '数字型') {
            if ($data[$field] <= 2147483647) $data[$field] += 0;
            else {
                $data[$field] = (double)$data[$field];//sprintf("%.0f", $data[$field]);

//                var_dump($data[$field]);//number_format($data[$field],0,'','');//
            }
        }
//        if (preg_match('#(Latitude|Longitude)#', $field) && $data[$field] < 180) {
//            $data[$field] = intval(1000000 * $data[$field]);
//        }
        if (strlen($data[$field]) > ($len = substr($info[2], 1))) {
            $errors[] = "{$field}: `{$data[$field]}` too long!{$len}";
        }
        if ('F' == substr($info[2], 0, 1) && strlen($data[$field]) != $len) {
            // if($len==8 && preg_match('#Date#', $field)){
            //     $data[$field] = '20160311';
            // }else
            $errors[] = "{$field}: length => $len !" . ($data[$field]);
        }
    }
    if ($errors) {
        return $errors;
    }

    if (isset($data[''])) unset($data['']);

    $data = [
        "Source"    => "0",
        "CompanyId" => 'syjcx',
        "IPCType"   => $IPCType,
        $IPCType    => $data
    ];

    return $data;
}

/**
 * post提交数据
 * @param $gateway
 * @param $req_data
 * @param array $headers
 * @param array $options
 * @return mixed
 */
function post($gateway, $req_data, $headers = [], $options = [])
{
    _log($gateway);
    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $gateway);
    curl_setopt($ch, CURLOPT_HEADER, 0);
    //HTTP头
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
    curl_setopt($ch, CURLOPT_POST, 1);
    curl_setopt($ch, CURLOPT_HTTPHEADER, array_merge([
        'Connection: keep-alive',
        'Content-Type: application/json',
        'Accept-Charset: utf-8',
    ], $headers));
    if ($options) {
        curl_setopt_array($ch, $options);
    }
    curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($req_data, JSON_UNESCAPED_SLASHES));
    $result = curl_exec($ch);
    _log($result);
    //_log(iconv('UTF-8', 'GBK', $result));
    $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
    curl_close($ch);
    return $httpCode;
}

/**
 * 上传文件
 * @param array $ftp config
 * @param $file
 * @param $upPath
 * @return int
 */
function upfile(array $ftp, $file, $upPath)
{
    $fileName = basename($file);
    if (!file_exists($file)) {
        echo '文件不存在！' . $file;
        return 404;
    }
    $remote_file = "/tmp/" . $fileName;

    $conn_id = ftp_ssl_connect($ftp['host']);

    $login_result = ftp_login($conn_id, $ftp['user_name'], $ftp['user_pass']) or print('ftp login failed!');

    ftp_pasv($conn_id, true);

    // upload file && move
    if (ftp_put($conn_id, $remote_file, $file, FTP_BINARY)
//        && ftp_raw($conn_id, "mv $remote_file swapfiles/$fileName")
        && ftp_rename_enhance($conn_id, $remote_file, "$upPath/$fileName")
    ) {
        $resultCode = 200;
        echo "successfully uploaded $file\n";

    } else {
        $resultCode = 500;
        echo "There was a problem while uploading $file\n";
    }
    ftp_close($conn_id);
    return $resultCode;
}

/**
 * ftp_name + 如果文件夹不存在则创建
 * @param $ftp_conn
 * @param $oldName
 * @param $newName
 * @return bool
 */
function ftp_rename_enhance($ftp_conn, $oldName, $newName)
{
    if (!ftp_rename($ftp_conn, $oldName, $newName)) {
        $ftpath = dirname($newName);
        $parts = explode('/', $ftpath);
        foreach ($parts as $part) {
            if (!@ftp_chdir($ftp_conn, $part)) {
                ftp_mkdir($ftp_conn, $part);
                ftp_chdir($ftp_conn, $part);
                ftp_chmod($ftp_conn, 0777, $part);
            }
        }
        return ftp_rename($ftp_conn, $oldName, $newName);
    }
    return true;
}

/**
 * 测试打印
 */
function _log()
{
    static $files = [];
    $bt = debug_backtrace();
    $caller = array_shift($bt);
    if (!isset($files[$caller['file']])) {
        $files[$caller['file']] = file($caller['file']);
    }

    preg_match('#\((.*)\)#i', $files[$caller['file']][$caller['line'] - 1], $params);

    echo "\nline:", $caller['line'], ": ", $params[1], "> ";

    if (func_num_args() > 1)
        print_r(func_get_args());
    else {
        $v = func_get_arg(0);
        if (is_array($v) || is_object($v)) var_dump($v);
        else echo $v;
    }
}

/**
 * 201188 => 20110808
 *
 * @param $date
 * @return int
 */
function guessDate($date)
{
    switch (strlen($date)) {
        case 6:
            $date = substr($date, 0, 4) . '0' . substr($date, 4, 1) . '0' . substr($date, 5, 1);
            break;
        case 7:
            $date = substr($date, 0, 4) . '0' . substr($date, 4, 3);
            break;
        case 8:
        default:
            ;
    }

    return intval($date);
}

function fileUp($file)
{
    return $file;
}

function unixTime($str)
{
    return strtotime($str);
}